package com.cristina.attention.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.cristina.attention.domain.MessageInfo;
import com.cristina.attention.domain.TaskInfo;
import com.cristina.attention.pending.ImTask;
import com.cristina.attention.pending.ImTaskPendingHolder;
import com.cristina.attention.pending.Task;
import com.cristina.attention.pending.TaskPendingHolder;
import com.cristina.attention.utils.GroupIdMappingUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * 消费 kafka 消息
 * @author by superman
 */
@Slf4j
public class Receiver {

    @Autowired
    private ApplicationContext context;

    @Resource
    private TaskPendingHolder taskPendingHolder;

    @Resource
    private ImTaskPendingHolder imTaskPendingHolder;

    /**
     * 监听用户的聊天, 保存到数据库, 推送公告 - (短信推送、邮箱推送...)  消费消息
     * @param consumerRecord
     * @param ack
     * @param topicGroupId
     */
    @KafkaListener(topics = "#{'${attention.topic.name}'}")
    public void consumer(ConsumerRecord<?, String> consumerRecord, Acknowledgment ack, @Header(KafkaHeaders.GROUP_ID) String topicGroupId) {
        Optional<String> kafkaMessage = Optional.ofNullable(consumerRecord.value());
        if (!kafkaMessage.isPresent()) return;

        TaskInfo taskInfo = JSON.parseObject(kafkaMessage.get(), TaskInfo.class);
        List<MessageInfo> messageInfoList = taskInfo.getMessageInfos();
        if (CollUtil.isEmpty(messageInfoList)) return;

        String messageGroupId = GroupIdMappingUtils.getGroupIdByMessageInfo(messageInfoList.get(0));
        if (!Objects.equals(topicGroupId, messageGroupId)) return;

        // 装配任务消息
        if (taskInfo.isImFlag()) {
            ImTask imTask = context.getBean(ImTask.class).setMessageInfo(messageInfoList.get(0));
            imTaskPendingHolder.route(topicGroupId).execute(imTask);
            ack.acknowledge();
            return;
        }
        for (MessageInfo messageInfo : messageInfoList) {
            Task imTask = context.getBean(Task.class).setMessageInfo(messageInfo);
            taskPendingHolder.route(topicGroupId).execute(imTask);
        }
        ack.acknowledge();
    }

}
